﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Protocol;
using MQTTnet.Server;
using Newtonsoft.Json;

namespace MqttBased
{
    public delegate Task<MqttMessage> ReceivedHander(byte[] input, Guid taskId);
    public class MqttClientProvider
    {
        private static MqttClient _mqttClient;
        private static ConcurrentDictionary<Guid, TaskCompletionSource<byte[]>> _taskHookInfos;
        private static ConcurrentDictionary<string, ReceivedHander> _TopicInfos;
        private static ConnState ConnState;

        public MqttClientProvider()
        {
            _taskHookInfos = _taskHookInfos ?? new ConcurrentDictionary<Guid, TaskCompletionSource<byte[]>>();
            _TopicInfos = _TopicInfos ?? new ConcurrentDictionary<string, ReceivedHander>();
        }

        public async Task CreateClientInstance(MqttSettingConfig setting,
            Action<MqttClient> registerHandler)
        {
            if (await CreateClientInstanceBase(setting))
            {
                registerHandler.Invoke(_mqttClient);
                Console.WriteLine("mqtt客户端启动成功!");
            }
            Task.Run(async () => await LoopConnection(setting));
        }
        public async Task CreateClientInstance(MqttSettingConfig setting)
        {
            if (await CreateClientInstanceBase(setting))
            {
                _mqttClient.Connected += (s, e) =>
                {
                    ConnState = ConnState.ConneSucc;
                    Console.WriteLine("已连接到MQTT服务器！");
                };
                _mqttClient.Disconnected += (s, e) =>
                {
                    ConnState = ConnState.ConnFail;
                    Console.WriteLine("已断开MQTT连接！");
                };
                Console.WriteLine("mqtt客户端启动成功!");
            }
            Task.Run(async () => await LoopConnection(setting));
        }

        private async Task LoopConnection(MqttSettingConfig setting)
        {
            while (true)
            {
                await Task.Delay(5000);
                if (ConnState == ConnState.ConnFail || ConnState == ConnState.None)
                {
                    if (await CreateClientInstanceBase(setting))
                    {
                        if (ConnState == ConnState.None)
                        {
                            _mqttClient.Connected += (s, e) =>
                            {
                                ConnState = ConnState.ConneSucc;
                                Console.WriteLine("已连接到MQTT服务器！");
                            };
                            _mqttClient.Disconnected += (s, e) =>
                            {
                                ConnState = ConnState.ConnFail;
                                Console.WriteLine("已断开MQTT连接！");
                            };
                            Console.WriteLine("mqtt客户端启动成功!");
                        }
                        ConnState = ConnState.ConneSucc;
                        foreach (var topic in _TopicInfos)
                        {
                            await SubscribeTopic(topic.Key, topic.Value);
                        }
                    }
                }
            }
        }
        public MqttClient GetClientInstance()
        {
            return _mqttClient;
        }
        public async Task CloseClient()
        {
            await _mqttClient.DisconnectAsync();
        }
        private async Task<bool> CreateClientInstanceBase(MqttSettingConfig setting)
        {
            try
            {
                if (new MqttFactory().CreateMqttClient() is MqttClient mqttClient)
                {
                    var clientOptions = new MqttClientOptionsBuilder()
                        //emq线上ssl版本
                        .WithTls(new MqttClientOptionsBuilderTlsParameters()
                        {
                            UseTls = true,
                            AllowUntrustedCertificates = true,
                            IgnoreCertificateChainErrors = true,
                            IgnoreCertificateRevocationErrors = true,
                            Certificates = new List<IEnumerable<byte>>()
                            {
                                new X509Certificate(@"F:\git\MqttMessageSample\ca.crt", setting.SslPassword).Export(X509ContentType.SerializedCert),
                                new X509Certificate(@"F:\git\MqttMessageSample\client.pfx", setting.SslPassword).Export(X509ContentType.SerializedCert),
                            }
                        })
                        //本地账号密码版本
                        //.WithCredentials(setting.UserName, setting.Password)
                        .WithTcpServer(setting.ServiceAddress, setting.Port)
                        .Build();
                    await mqttClient.ConnectAsync(clientOptions);
                    _mqttClient = mqttClient;
                    _mqttClient.ApplicationMessageReceived += MqttClient_ApplicationMessageReceived;
                }
                else
                {
                    Console.WriteLine("客户端实例创建失败!");
                    return false;
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
                return false;
            }
            return true;
        }


        private async void MqttClient_ApplicationMessageReceived(object sender,
            MqttApplicationMessageReceivedEventArgs e)
        {
            var message =
                JsonConvert.DeserializeObject<MqttMessage>(
                    Encoding.UTF8.GetString(e.ApplicationMessage.Payload));
            var inputBody = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message.MessageBody));
            switch (message.MessageType)
            {
                case MessageType.Send:
                    Console.WriteLine($"当前收到的消息：>> {JsonConvert.SerializeObject(message.MessageBody)}");
                    if (_TopicInfos.Any(x => x.Key.Equals(e.ApplicationMessage.Topic)))
                    {
                        var topic = _TopicInfos.FirstOrDefault(x => x.Key.Equals(e.ApplicationMessage.Topic));
                        if (topic.Value != null)
                        {
                            if (message.NeedReceived)
                            {
                                var messageBody = await topic.Value(inputBody, message.TaskId);
                                if (_mqttClient != null && _mqttClient.IsConnected)
                                {
                                    await _mqttClient.PublishAsync(new MqttApplicationMessage
                                    {
                                        Topic = message.ReceivedTopicName,
                                        Payload = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(messageBody))
                                    });
                                    Console.WriteLine($"回调处理事件完成,回调消息：>> {JsonConvert.SerializeObject(messageBody)}");
                                }
                                else
                                {
                                    //todo 可接入polly实现消息重试提高通讯健壮性
                                }
                            }
                            else
                            {
                                await topic.Value(inputBody, message.TaskId);
                                Console.WriteLine($"处理事件完成,结束");
                            }
                        }
                        else
                        {
                            Console.WriteLine($"没有注册回调处理事件,直接结束!");
                        }
                    }
                    break;
                case MessageType.Back:
                    Console.WriteLine($"当前收到的消息：>> {JsonConvert.SerializeObject(message.MessageBody)}");
                    if (_taskHookInfos.Any(x => x.Key == message.TaskId))
                    {
                        var myhook = _taskHookInfos.FirstOrDefault(x => x.Key == message.TaskId);
                        myhook.Value.TrySetResult(inputBody);
                    }
                    break;
            }
        }
        /// <summary>
        /// 获取回调信息
        /// </summary>
        /// <returns></returns>
        public TaskCompletionSource<byte[]> GetHook(Guid id)
        {
            var task = _taskHookInfos.FirstOrDefault(x=>x.Key.Equals(id));
            return task.Value;
        }
        /// <summary>
        /// 删除钩子
        /// </summary>
        /// <param name="id"></param>
        public void RemoveHook(Guid id)
        {
            _taskHookInfos.TryRemove(id, out TaskCompletionSource<byte[]> task);
        }

        /// <summary>
        /// 设置注册钩子
        /// </summary>
        public void SetHook(Guid id, TaskCompletionSource<byte[]> message)
        {
            _taskHookInfos.TryAdd(id, message);
        }
        public async Task SubscribeTopic(string topicName, ReceivedHander hander)
        {
            if (string.IsNullOrEmpty(topicName))
            {
                Console.WriteLine("订阅主题不能为空！");
                return;
            }
            _TopicInfos.TryAdd(topicName, hander);
            if (_mqttClient != null)
            {
                if (!_mqttClient.IsConnected)
                {
                    Console.WriteLine("MQTT客户端尚未连接！");
                    return;
                }
                await _mqttClient.SubscribeAsync(new List<TopicFilter>
                {
                    new TopicFilter(topicName, MqttQualityOfServiceLevel.AtMostOnce)
                });
                Console.WriteLine($"已订阅[{topicName}]主题");
            }
        }

        public async Task<TOut> SendMessage<TOut>(MqttMessage message)
        {
            if (string.IsNullOrEmpty(message.TopicName))
            {
                Console.WriteLine("发布主题不能为空！");
                return default(TOut);
            }
            if (_mqttClient == null || !_mqttClient.IsConnected)
            {
                Console.WriteLine("MQTT客户端尚未连接！");
                return default(TOut);
            }
            var appMsg = new MqttApplicationMessage
            {
                Topic = message.TopicName,
                Payload = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message))
            };
            TaskCompletionSource<byte[]> task;
            if (message.NeedReceived)
            {
                task = new TaskCompletionSource<byte[]>();
                SetHook(message.TaskId, task);
            }
            await _mqttClient.PublishAsync(appMsg);
            if (message.NeedReceived)
            {
                byte[] result = default(byte[]);
                var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
                task = GetHook(message.TaskId);
                await task.WaitAsync(cts.Token);
                if (cts.IsCancellationRequested)
                {
                    Console.WriteLine($"超时5秒，取消等待");
                    task.TrySetCanceled();
                }
                else
                {
                    result = task.Task.Result;
                    return JsonConvert.DeserializeObject<TOut>(Encoding.UTF8.GetString(result));
                }
                RemoveHook(message.TaskId);
            }
            return default(TOut);
        }
    }
}
